/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.listeners;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import br.pucrio.inf.lac.konoha.hyuga.common.Grid;
import br.pucrio.inf.lac.konoha.hyuga.core.DDSLink;
import br.pucrio.inf.lac.konoha.hyuga.core.HyugaService;
import br.pucrio.inf.lac.konoha.hyuga.events.CEPEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellDisperseEvent;

import com.espertech.esper.client.EPException;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.UpdateListener;

public class CellOutputListener {
    
    /** Hyuga. */
    protected HyugaService hyuga;
    
    /** DDSLink (Data Stream). */
    protected DDSLink ddsLink;

    /** Thread Pool. */
    protected ExecutorService controllerPool;
    
    /** LOGGER. */
    protected static Logger LOGGER = Logger.getLogger(CellOutputListener.class.getCanonicalName());
    
    /**
     * Default Constructor. Receive the DDSLink to send updates to other instances or management apps.
     * @param ddsLink - DDSLink
     */
    public CellOutputListener(HyugaService hyuga, DDSLink ddsLink) {
        this.hyuga   = hyuga;
        this.ddsLink = ddsLink;
        
        if (hyuga.isLocalDeploy()) {
            controllerPool = Executors.newFixedThreadPool(16);
        } else {
            controllerPool = Executors.newFixedThreadPool(16);
        }
    }

    public void update(CellClusterEvent cellCluster) {
        controllerPool.submit(new GridThread(cellCluster));
    }
    
    public void update(CellDisperseEvent cellDisperse) {
        controllerPool.submit(new GridThread(cellDisperse));
    }

    public class GridThread implements Runnable {
        CEPEvent cepEvent;
        
        public GridThread(CEPEvent cepEvent) {
            this.cepEvent = cepEvent;
        }

        @Override
        public void run() {
            int x = 0;
            int y = 0;

            if (cepEvent instanceof CellClusterEvent) {
                ((CellClusterEvent) cepEvent).setDistributed(true);
                x = ((CellClusterEvent) cepEvent).getCore().getX();
                y = ((CellClusterEvent) cepEvent).getCore().getY();

                Grid.gridReadLock();
            }
            
            if (cepEvent instanceof CellDisperseEvent) {
                ((CellDisperseEvent) cepEvent).setDistributed(true);
                x = ((CellDisperseEvent) cepEvent).getX();
                y = ((CellDisperseEvent) cepEvent).getY();
            }
            
            
            double lat = hyuga.getLatitudeSegments().getSegments()[y];
            double lng = hyuga.getLongitudeSegments().getSegments()[x];

            try {
                if (hyuga.isLocalDeploy()) {
                    Grid.cellLock(x, y);
                    
                    hyuga.getCEPEngine().getEPRuntime().sendEvent(cepEvent);
                    
                    Grid.cellUnlock(x, y);
                } else {
                    ddsLink.publishSpatial("GridEvent",
                                           lat * 1000000,
                                           lng * 1000000,
                                           cepEvent.getType().toString(),
                                           cepEvent, 
                                           cepEvent.getType().toString());
                    LOGGER.info("Published Grid Event: " + cepEvent.getType().toString());
                }
            } catch (EPException ex) {
                LOGGER.error("Exception at processing cellcluster/disp event" + ex.getMessage());
            }
        }
        
    }
}
